package cn.me.alphamq.nameserver.event.spi.listener.cluster.master;

import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.util.NettyUtil;
import cn.me.alphamq.nameserver.cache.Container;
import cn.me.alphamq.nameserver.event.model.cluster.StartReplicationReqEvent;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.nameserver.util.NameServerUtil;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/**
 * 主从复制中用于主节点的监听器，监听开启同步复制的请求事件
 *
 * @author f
 */
public class MasterStartReplicationListener implements Listener<StartReplicationReqEvent> {

    private static final Logger log = LoggerFactory.getLogger(MasterStartReplicationListener.class);

    @Override
    public void onReceive(StartReplicationReqEvent event) throws Exception {
        ChannelHandlerContext ctx = event.getCtx();

        // 如果身份校验不通过，则报错
        if (!NameServerUtil.verifyAuth(event.getUsername(), event.getPassword())) {
            ctx.writeAndFlush(new TcpMsg(EventCodeEnum.ERROR_ACCOUNT_RESP));
            ctx.close();
            throw new IllegalAccessException("incorrect username or password");
        }

        // 保存从节点
        InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
        String ctxId = inetSocketAddress.getHostName() + ":" + inetSocketAddress.getPort();
        NettyUtil.setCtxId(ctx, ctxId);
        Container.getReplicationChannelManager().put(ctxId, ctx);

        log.info("slave [{}] start replication...", ctxId);

        // 给从节点响应消息
        ctx.writeAndFlush(new TcpMsg(EventCodeEnum.MASTER_START_REPLICATION_RESP));
    }
}
